package com.heiye.comment.biz.consumer;

import cn.hutool.core.collection.CollUtil;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.RateLimiter;
import com.heiye.comment.biz.constant.MQConstants;
import com.heiye.comment.biz.domain.dataobject.CommentDO;
import com.heiye.comment.biz.domain.mapper.CommentDOMapper;
import com.heiye.comment.biz.model.dto.DeleteCommentCacheMQDTO;
import com.heiye.comment.biz.model.vo.DeleteCommentReqVO;
import com.heiye.comment.biz.rpc.CountRpcService;
import com.heiye.comment.biz.rpc.KeyValueRpcService;
import com.heiye.framework.common.constant.DateConstants;
import com.heiye.framework.common.util.JsonUtils;
import com.heiye.kv.api.dto.req.FindCommentContentReqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.concurrent.CompletableFuture;

/**
 * @author: heiye
 * @date: 2025/08/04 下午8:15
 * @version: v1.0.0
 * @description: 删除评论 - 后续业务处理
 */
@Slf4j
@Component
@RocketMQMessageListener(
        consumerGroup = "interact_group_" + MQConstants.TOPIC_DELETE_COMMENT,
        topic = MQConstants.TOPIC_DELETE_COMMENT
)
public class DeleteCommentConsumer implements RocketMQListener<String> {

    @Resource
    private CommentDOMapper commentDOMapper;

    @Resource
    private KeyValueRpcService keyValueRpcService;

    @Resource
    private CountRpcService countRpcService;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Resource(name = "taskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    // 每秒创建 1000 个令牌
    private RateLimiter rateLimiter = RateLimiter.create(1000);

    @Override
    public void onMessage(String body) {
        // 令牌桶流控
        rateLimiter.acquire();

        log.info("## 【删除评论 - 后续业务处理】消费者消费成功, body: {}", body);

        CommentDO commentDO = JsonUtils.parseObject(body, CommentDO.class);

        Long commentId = commentDO.getId();

        // 1. 批量删除关联评论（递归查询回复评论，并批量删除）
        List<Long> replyCommentIds = Lists.newArrayList();
        List<FindCommentContentReqDTO> findCommentContentReqDTOS = Lists.newArrayList();
        recurrentGetReplyCommentId(replyCommentIds, findCommentContentReqDTOS, List.of(commentId));

        // 编程式事务，保证原子性
        transactionTemplate.execute(status -> {
            try {

                // 被删除的行数
                int count = 0;
                // RPC: 调用 kv 键值服务
                CompletableFuture<Boolean> KVFuture = CompletableFuture.completedFuture(null);

                // 当回复评论不为空时，才批量删除关联评论和其元数据
                if (CollUtil.isNotEmpty(replyCommentIds)) {
                    count = commentDOMapper.deleteByIds(replyCommentIds);

                    KVFuture = CompletableFuture.supplyAsync(() ->
                            keyValueRpcService.batchDeleteCommentContent(commentDO.getQuizId(), findCommentContentReqDTOS), threadPoolTaskExecutor);
                }

                // RPC: 调用计数服务，计数更新（问题下总评论数）
                int finalCount = -(count + 1);
                CompletableFuture<Boolean> countFuture = CompletableFuture.supplyAsync(() ->
                        countRpcService.updateQuizCommentCount(commentDO.getQuizId(), finalCount), threadPoolTaskExecutor);

                CompletableFuture.allOf(KVFuture, countFuture).join();
                return true;
            } catch (Exception ex) {
                // 标记事务为回滚
                status.setRollbackOnly();
                log.error("", ex);
                throw ex;
            }
        });

        // 如果回复评论为空，则返回无需删除缓存
        if (CollUtil.isEmpty(replyCommentIds)) {
            return;
        }

        // 2. 发送 MQ 批量删除关联评论缓存
        DeleteCommentCacheMQDTO deleteCommentCacheMQDTO = DeleteCommentCacheMQDTO.builder()
                .quizId(commentDO.getQuizId())
                .commentIds(replyCommentIds)
                .build();

        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(deleteCommentCacheMQDTO))
                .build();

        rocketMQTemplate.asyncSend(MQConstants.TOPIC_DELETE_COMMENT_CACHE, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【评论缓存删除】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【评论缓存删除】MQ 发送失败，发送异常：", throwable);
            }
        });

    }

    /**
     * 递归获取全部回复的评论 ID
     *
     * @param commentIds
     * @param findCommentContentReqDTOS
     * @param needQueryCommentIds
     */
    private void recurrentGetReplyCommentId(List<Long> commentIds, List<FindCommentContentReqDTO> findCommentContentReqDTOS, List<Long> needQueryCommentIds) {
        List<CommentDO> replyCommentList = commentDOMapper.selectByReplyCommentId(needQueryCommentIds);

        if (CollUtil.isEmpty(replyCommentList)) {
            return;
        }

        List<Long> replyCommentIds = replyCommentList.stream().map(CommentDO::getId)
                // 去重
                .distinct()
                .toList();

        List<FindCommentContentReqDTO> findCommentContentReqDTOList = replyCommentList.stream()
                .map(replyComment -> FindCommentContentReqDTO.builder()
                        .contentId(replyComment.getContentUuid())
                        .yearMonth(DateConstants.DATE_FORMAT_Y_M.format(replyComment.getCreateTime()))
                        .build()
                )
                .toList();

        commentIds.addAll(replyCommentIds);
        findCommentContentReqDTOS.addAll(findCommentContentReqDTOList);

        // 递归调用
        recurrentGetReplyCommentId(commentIds, findCommentContentReqDTOS, replyCommentIds);
    }
}
